package com.zfzn.flink;

import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @Auther lolineet
 * @Email tj.zhanghongwei@163.com
 * @Date 2021/8/19 10:27
 * @Description
 */
public class FraudDetector extends KeyedProcessFunction<Long,Transaction,Alert> {

    private static final double SMALL_AMOUNT = 1.00;
    private static final double LARGE_AMOUNT = 500.00;
    private static final long ONE_MINUTE = 60 * 1000;

    private transient ValueState<Boolean> flagState;
    private transient ValueState<Long> timerState;
    public void open(Configuration parameters){
        ValueStateDescriptor<Boolean> flagDescriptor=new ValueStateDescriptor<Boolean>("flag", Types.BOOLEAN);
        flagState = getRuntimeContext().getState(flagDescriptor);

        ValueStateDescriptor<Long> timerDescriptor = new ValueStateDescriptor<Long>("timer-state",Types.LONG);
        timerState = getRuntimeContext().getState(timerDescriptor);
    }

    @Override
    public void processElement(Transaction transaction, Context context, Collector<Alert> collector) throws Exception {
        Boolean lastAmountWasSmall= flagState.value();
        if(lastAmountWasSmall!=null){
            if(transaction.getAmount()>LARGE_AMOUNT){
                Alert alert = new Alert();
                alert.setId(transaction.getAccountId());
                alert.setAmount(transaction.getAmount());
                collector.collect(alert);
            }
            flagState.clear();
        }
        if(transaction.getAmount()<SMALL_AMOUNT){
            flagState.update(true);
            long timer = context.timerService().currentProcessingTime()+ONE_MINUTE;
            context.timerService().registerEventTimeTimer(timer);
            timerState.update(timer);
        }
    }

    @Override
    public void onTimer(long timestamp,OnTimerContext ctx,Collector<Alert> out){
        timerState.clear();
        flagState.clear();
    }
}
